package club.drguo.mapreduce.inverseindex;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
//club.drguo.mapreduce.inverseindex.InverseIndexStepTwo
public class InverseIndexStepTwo {
	// K：起始偏移量 V： {hello--a.txt	3}
	public static class InverseIndexStepTwoMapper extends Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			//切分出各個字段
			String[] strings = StringUtils.split(line, "\t");
			
			String wordAndFile = strings[0];
			long count = Long.parseLong(strings[1]);
			String[] wordAndFileName = StringUtils.split(wordAndFile, "--");
			
			String word = wordAndFileName[0];
			String fileName = wordAndFileName[1];
			//將單詞作爲key，文件-->次數作爲value輸出 形式：<hello, a.txt--3>
			k.set(word);
			v.set(fileName+"--"+count);
			context.write(k, v);
		}
	}
	public static class InverseIndexStepTwoReducer extends Reducer<Text, Text, Text, Text>{
		@Override
		protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			//拿到的数据<hello,{a.txt--3,b.txt--2,...}>
			String result = "";
			for(Text value : values){
				result += value+" ";
			}
			context.write(key, new Text(result));//输出结果 K：hello V：a.txt--3 b.txt--2 ...
		}
	}
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job_stepTwo = Job.getInstance(conf);

		job_stepTwo.setJarByClass(InverseIndexStepTwo.class);

		job_stepTwo.setMapperClass(InverseIndexStepTwoMapper.class);
		job_stepTwo.setReducerClass(InverseIndexStepTwoReducer.class);

		job_stepTwo.setOutputKeyClass(Text.class);
		job_stepTwo.setOutputValueClass(Text.class);

		FileInputFormat.setInputPaths(job_stepTwo, new Path(args[0]));
		// 先判断一下，如果输出目录已存在则删除
		FileSystem fileSystem = FileSystem.get(conf);
		Path output = new Path(args[1]);
		if (fileSystem.exists(output)) {
			fileSystem.delete(output, true);// true递归删除
		}
		FileOutputFormat.setOutputPath(job_stepTwo, output);

		System.exit(job_stepTwo.waitForCompletion(true) ? 0 : 1);

	}
}
